package com.lagou.no3

import java.text.SimpleDateFormat

import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment, WindowedStream}
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.util.Collector


object TumblingWindowDemo {
    def main(args: Array[String]): Unit = {

        //获取流环境
        val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        //设置并行度
        env.setParallelism(1)
        //自定义数据源生成数据，每秒生成一条当前时间戳
        val dataStream: KeyedStream[(String, Long), String] = env.addSource(new SourceFunction[(String, Long)] {
            var flag = true

            override def run(sourceContext: SourceFunction.SourceContext[(String, Long)]): Unit = {
                while (flag) {
                    sourceContext.collect(("hello", System.currentTimeMillis()))
                    Thread.sleep(1000)

                }
            }

            override def cancel(): Unit = {
                flag = false;
            }
        }).keyBy(_._1)
        //将key流转为窗口流，设置窗口长度为10秒
        val windowStream: WindowedStream[(String, Long), String, TimeWindow]
            = dataStream.timeWindow(Time.seconds(10))
        //调用apply方法转为DataStream，重写windowFunction方法，获取窗口中的数据
        val res: DataStream[(String, Long)] = windowStream.apply(new WindowFunction[(String, Long), (String, Long), String, TimeWindow] {
            override def apply(key: String, window: TimeWindow, input: Iterable[(String, Long)], out: Collector[(String, Long)]): Unit = {
                val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")
                input.foreach(tup => out.collect((s"${tup._1},开始时间:${sdf.format(window.getStart)},结束时间:${sdf.format(window.getEnd)},value:${sdf.format(tup._2)}", tup._2)))
                println("=================================================本次窗口结束====================================================================")
            }
        })
        //打印结果
        res.print()

        env.execute()



    }

}

